package com.xzc.kafka.action.producer;

import com.alibaba.fastjson.JSON;
import com.xzc.kafka.action.pojo.User;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Map;
import java.util.Properties;

public class _02Serializer implements Serializer {


    @Override
    public void configure(Map configs, boolean isKey) {
        System.out.println(1);
    }

    @Override
    public byte[] serialize(String topic, Object data) {
        byte[] ret = null;
        try {
//            ret = objectMapper.writeValueAsString(data).getBytes("UTF-8");
            ret = JSON.toJSONBytes(data);
            System.out.println(2);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ret;
    }

    @Override
    public void close() {

    }

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", "es1:19092,es2:19092,es3:19092");
//        props.put("key.serializer", "org.apache.kafka.common.serializtion.StringSerializer");
//        props.put("value.serializer", "org.apache.kafka.common.serializtion.StringSerializer");
        props.put("acks", "-1");
        props.put("retries", 3);
        props.put("batch.size", 323840);
        props.put("linger.ms", 10);
        props.put("buffer.memory", 33554432);
        props.put("max.block.ms", 3000);

//        props.put("compression.type", "lz4");

        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.xzc.kafka.action.producer._01AuditPartitioner");

        Serializer<String> keySerl = new _02Serializer();
        Serializer<User> valSerl = new _02Serializer();
        Producer<String, User> producer = new KafkaProducer<String, User>(props, new StringSerializer(), valSerl);

        String topic = "test-topic1";
        User user = new User("X", "ZC", 36, "WUHAN");
        ProducerRecord record = new ProducerRecord(topic, user.toString(), user);
//        producer.send(record).get();
        sendAndCallback(producer, record);
//
        producer.close();
    }

    public static void sendAndCallback(Producer<String, User> producer, ProducerRecord record) {
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception == null) {
                    System.out.println("消息收到了" + metadata.offset());
                } else {
                    if (exception instanceof RetriableException) {
                        System.out.println("消息异常了，但可以重试，重试可能引起消息时序混乱");
                    } else {
                        System.out.println("消息异常了，不可重试");
                    }
                }
            }
        });
    }
}
